package com.atguigu.userprofile.app

import com.atguigu.userprofile.bean
import com.atguigu.userprofile.bean.TagInfo
import com.atguigu.userprofile.constant.CodeConst
import com.atguigu.userprofile.dao.TagInfoDAO
import com.atguigu.userprofile.util.MyClickhouseUtil
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ListBuffer

object TaskBitmapApp {


  //特别说明： 拼接sql driver  执行sql clickhouse  没有用到ex
  // 其实并不是一个spark程序，也并不一定需要yarn分配资源
  // 但是为了所有应用的统一管理 依然把该程序做成spark程序 提交yarn

  //1 从mysql取得标签定义集合
  //

  //2  组合一条 insert select 操作
  //从clickhouse宽表读取
  //
  //写入clickhouse的bitmap表
  def main(args: Array[String]): Unit = {
    val taskId: String = args(0)
    val busiDate: String = args(1)
    val sparkConf= new SparkConf().setAppName("task_bitmap_app")//.setMaster("local[*]")
    val sparkContext = new SparkContext(sparkConf)

    //1 查询已启用的标签列表
    val tagInfoList: List[bean.TagInfo] = TagInfoDAO.getTagInfoListWithOn()


    //2 因为要按标签值的类型不同，分成四份，分别插入4种bitmap表中
    val tagInfoStringList = new ListBuffer[TagInfo]()
    val tagInfoLongList = new ListBuffer[TagInfo]()
    val tagInfoDecimalList = new ListBuffer[TagInfo]()
    val tagInfoDateList = new ListBuffer[TagInfo]()

    //2.1分成四份
    for (tagInfo <- tagInfoList) {
      if (tagInfo.tagValueType == CodeConst.TAG_VALUE_TYPE_STRING) {
        tagInfoStringList.append(tagInfo)
      } else if (tagInfo.tagValueType == CodeConst.TAG_VALUE_TYPE_LONG) {
        tagInfoLongList.append(tagInfo)
      } else if (tagInfo.tagValueType == CodeConst.TAG_VALUE_TYPE_DECIMAL) {
        tagInfoDecimalList.append(tagInfo)
      } else if (tagInfo.tagValueType == CodeConst.TAG_VALUE_TYPE_DATE) {
        tagInfoDateList.append(tagInfo)
      }

    }
    //2.2 把4个list 转换为四条sql 分别执行
    insertBitmap(tagInfoStringList, "user_tag_value_string", busiDate)
    insertBitmap(tagInfoLongList, "user_tag_value_long", busiDate)
    insertBitmap(tagInfoDecimalList, "user_tag_value_decimal", busiDate)
    insertBitmap(tagInfoDateList, "user_tag_value_date", busiDate)


  }


  //跟list插入到目标bitmap表
  def insertBitmap(tagInfoList: ListBuffer[TagInfo], targetTableName: String, busiDate: String): Unit = {

    if (tagInfoList.size > 0) {
      val tagCodeList: ListBuffer[String] = tagInfoList.map(tagInfo => s"('${tagInfo.tagCode.toLowerCase}',${tagInfo.tagCode.toLowerCase})")
      val tagCodeSQL: String = tagCodeList.mkString(",")

      //幂等性处理
      val deleteSQL=s" alter table $targetTableName delete where dt='$busiDate'"
      println(deleteSQL)
      MyClickhouseUtil.executeSql(deleteSQL);


      val sql =
        s"""
           |insert into  $targetTableName
           |select  tt.tg.1 as tag_code,if(tt.tg.2='','0',tt.tg.2) as tag_value , groupBitmapState(cast (uid as UInt64)  ) ,'$busiDate'
           |from
           |(
           |select uid ,arrayJoin([$tagCodeSQL]) tg
           |from up_tag_merge_${busiDate.replace("-", "")}
           |) tt
           |group by  tt.tg.1 as tag_code,tt.tg.2
           |""".stripMargin

      println(sql)

      MyClickhouseUtil.executeSql(sql);
    }

  }

}
